package com.zy.gmall.realtime.app.func;


import com.alibaba.fastjson.JSONObject;
import com.zy.gmall.realtime.bean.TableProcess;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.*;

public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, TableProcess, JSONObject> {

    private MapStateDescriptor<String,TableProcess> mapStateDescriptor;

    private Map<String,TableProcess> mapConfig = new HashMap<>();


    public TableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor){
        this.mapStateDescriptor =  mapStateDescriptor;
    }


    //！！！！注意一个点 当出问题重启的时候不知道主流和广播流哪个先  解决：预处理
    @Override
    public void open(Configuration parameters) throws Exception {
        //注册驱动
        Class.forName("com.mysql.cj.jdbc.Driver");
        //建立连接
        Connection conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/gmall_config?" +
                "user=root&password=000000&useUnicode=true&characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");
        //获取数据库对象
        String sql = "select * from gmall_config.table_process_dim ";
        PreparedStatement ps = conn.prepareStatement(sql);
        //执行SQL语句
        ResultSet rs = ps.executeQuery();
        //处理结果
        ResultSetMetaData metaData = rs.getMetaData();
        while (rs.next()){
            JSONObject jsonObject = new JSONObject();
            for (int i = 1; i < metaData.getColumnCount(); i++) {
                String columnName = metaData.getColumnName(i);
                Object rsObject = rs.getObject(i);
                jsonObject.put(columnName, rsObject);
            }
            //将json转换成实体类
            TableProcess tableProcess = jsonObject.toJavaObject(TableProcess.class);
            //放入mapConfig
            mapConfig.put(tableProcess.getSourceTable(), tableProcess);
        }
        //释放资源
        rs.close();
        ps.close();
        conn.close();
    }

    //处理主流数据
    @Override
    public void processElement(JSONObject jsonObject, ReadOnlyContext readOnlyContext, Collector<JSONObject> collector) throws Exception {
        // {"database":"gmall0315","data":{"tm_name":"苹果11","create_time":"2021-12-14 00:00:00","logo_url":"/static/default.jpg2323","id":2},
        // "old":{"tm_name":"苹果"},"type":"update","table":"base_trademark","ts":1692585633}
        // System.out.println(jsonObj);
        //获取处理业务数据表表名
        String table = jsonObject.getString("table");
        //获取广播状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);
        //根据业务数据库表表名到广播状态中获取对应的配置信息
        TableProcess tableProcess = null;
        tableProcess = broadcastState.get(table);
        //判断配置信息是否存在
        if (tableProcess != null ||(tableProcess = mapConfig.get(table)) != null){
            //不为空，则存在于广播状态中
            //{"tm_name":"苹果1122","create_time":"2021-12-14 00:00:00","logo_url":"/static/default.jpg2323","id":2}
            //只拿到date数据信息 其他用不到
            JSONObject objectData = jsonObject.getJSONObject("data");

            //过滤一下 data中只要id tm_name
            //全部列
            String sinkColumns = tableProcess.getSinkColumns();
            filterColumn(objectData,sinkColumns);

            //进行修改 操作信息
            String type = jsonObject.getString("type");
            objectData.put("type",type);

            //表信息
            objectData.put("table_process",tableProcess);
            //向下游发送
            collector.collect(objectData);
        }
    }

    private void filterColumn(JSONObject objectData, String sinkColumns) {
        //过滤思路：
        //有用列名
        String[] columns = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columns);
        //所有列名
        Set<Map.Entry<String, Object>> entrySet = objectData.entrySet();


        entrySet.removeIf(entry-> !columnList.contains(entry.getKey()));
//        for (Map.Entry<String, Object> entry : entries) {
//            //遍历所有列名 判断如果不在有用中就删除
//            if (!column.contains(entry.getKey())){
//                entries.remove(entry);
//            }
//        }
    }

    //处理广播数据 表信息
    @Override
    public void processBroadcastElement(TableProcess tableProcess, BroadcastProcessFunction<JSONObject, TableProcess, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
        //获取广播状态
        BroadcastState<String, TableProcess> broadcastState = context.getBroadcastState(mapStateDescriptor);
        //获取类型
        String op = tableProcess.getOp();
        //获取表名
        String sourceTable = tableProcess.getSourceTable();
        //判断是否是删除操作
        if ("d".equals(op)){
            //将对应的配置信息从广播状态中删除掉
            broadcastState.remove(sourceTable);
            mapConfig.remove(sourceTable);
        }else {
            //如果对配置表做了删除外的其它操作，将对应的配置信息放到或者更新到广播状态中
            broadcastState.put(sourceTable, tableProcess);
            mapConfig.put(sourceTable,tableProcess);
        }

    }
}
